package com.hkbigdata.flink.task;


import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.hkbigdata.flink.domain.ProductProcess;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.producer.ProducerConfig;

import static com.hkbigdata.flink.utils.Constants.*;

public class C02_ProductWarning {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        // TODO 从Kafka读：新Source架构
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers(KAFKA_BOOSTRAP_SERVERS) // 指定kafka节点的地址和端口
//                .setGroupId("order")  // 指定消费者组的id
                .setTopics(KAFKA_TOPIC_CDC)   // 指定消费的 Topic
                .setValueOnlyDeserializer(new SimpleStringSchema()) // 指定 反序列化器，这个是反序列化value
                .setStartingOffsets(OffsetsInitializer.earliest())  // flink消费kafka的策略
                .build();


        SingleOutputStreamOperator<String> proDs = env
                .fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafkasource")
                .flatMap(new FlatMapFunction<String, ProductProcess>() {
                    @Override
                    public void flatMap(String value, Collector<ProductProcess> out) throws Exception {
                        //System.out.println(value);
                        try {
                            ObjectMapper objectMapper = new ObjectMapper();
                            JsonNode jsonNode = objectMapper.readTree(value);
                            ProductProcess productProcess = new ProductProcess();
                            String table = jsonNode.get("source").get("table").asText();
                            if (!"prd_product".equals(table)) {
                                return;
                            }
                            Long id = jsonNode.get("after").get("id").asLong();
                            String name = jsonNode.get("after").get("name").asText();
                            Long stock = jsonNode.get("after").get("stock").asLong();
                            Long lowStock = jsonNode.get("after").get("low_stock").asLong();

                            productProcess.setId(id);
                            productProcess.setName(name);
                            productProcess.setStock(stock);
                            productProcess.setLowStack(lowStock);
                            System.out.println("数据2=" + productProcess);
                            out.collect(productProcess);

                        } catch (Exception e) {
                            System.out.println("异常为：" + e.getMessage() + " 异常数据为：" + value);
                        }
                    }
                }).filter(new FilterFunction<ProductProcess>() {
                    @Override
                    public boolean filter(ProductProcess value) throws Exception {
                        return value.getStock() == value.getLowStack();
                    }
                }).process(new ProcessFunction<ProductProcess, String>() {
                    @Override
                    public void processElement(ProductProcess value, ProcessFunction<ProductProcess, String>.Context ctx, Collector<String> out) throws Exception {
                        StringBuilder stringBuilder = new StringBuilder();
                        stringBuilder.append(value.getId() + ":").append(value.getName() + ":").append(value.getStock() + ":").append(value.getLowStack());
                        // 数据处理好后发给下游
                        out.collect(stringBuilder.toString());
                    }
                });

        proDs.print();

        KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
                // 指定 kafka 的地址和端口
                .setBootstrapServers(KAFKA_BOOSTRAP_SERVERS)
                // 指定序列化器：指定Topic名称、具体的序列化
                .setRecordSerializer(
                        KafkaRecordSerializationSchema.<String>builder()
                                .setTopic("product_warning")
                                .setValueSerializationSchema(new SimpleStringSchema())
                                .build()
                )
                // 写到kafka的一致性级别： 精准一次、至少一次
                .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
                // 如果是精准一次，必须设置 事务的前缀
                .setTransactionalIdPrefix("product-process3-")
                // 如果是精准一次，必须设置 事务超时时间: 大于checkpoint间隔，小于 max 15分钟
                .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 10 * 60 * 1000 + "")
                .build();

        proDs.sinkTo(kafkaSink);
        env.execute();
    }
}
